package com.shujia.flink.sql;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class Demo02WordCntSQL {
    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                // 指定为流处理的运行模式
                .inStreamingMode()
                // 指定为批处理的运行模式
                //.inBatchMode()
                .build();

        // 从Kafka消费student数据，统计班级人数

        // 第二种方式构建Table环境
        TableEnvironment tEnv = TableEnvironment.create(settings);

        // 创建Source
        tEnv.executeSql("CREATE TABLE students_kafka (\n" +
                "    sid STRING,\n" +
                "    name STRING,\n" +
                "    age INT,\n" +
                "    sex STRING,\n" +
                "    clazz STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'student01', -- 数据的topic\n" +
                "  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表\n" +
                "  'properties.group.id' = 'testGroup', -- 消费者组\n" +
                "  'scan.startup.mode' = 'latest-offset', -- 读取数据的位置earliest-offset latest-offset\n" +
                "  'format' = 'csv' -- 读取数据的格式\n" +
                ");");

        // 创建Sink
        tEnv.executeSql("CREATE TABLE clazz_cnt (\n" +
                "    clazz STRING,\n" +
                "    cnt BIGINT\n" +
                ") WITH (\n" +
                "     'connector' = 'print'\n" +
                ");");

        // 执行插入
        tEnv.executeSql("insert into clazz_cnt\n" +
                "select clazz\n" +
                "        ,count(*) as cnt\n" +
                "from students_kafka\n" +
                "group by clazz;");


    }
}
